﻿using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;

namespace Ec.NETCoreFramework.Common.MQ
{
    public class MsgConsume1
    {
        private IConnection _connection = null;
        private string _hostName = string.Empty;
        private string _userName = string.Empty;
        private string _pwd = string.Empty;
        private string _virtualHost = string.Empty;
        public MsgConsume1(string hostName, string userName, string pwd, string virtualHost)
        {
            _hostName = hostName;
            _userName = userName;
            _pwd = pwd;
            _virtualHost = virtualHost;
            var factory = new ConnectionFactory()
            {
                HostName = _hostName,
                Password = _pwd,
                UserName = _userName,
                VirtualHost = _virtualHost
            };
            factory.AutomaticRecoveryEnabled = true;
            _connection = factory.CreateConnection();
        }
        public void Consume(TaskInterface consumeInterface)
        {
            Task.Run(() =>
            {
                var taskInterface = consumeInterface;
                using (var channel = _connection.CreateModel())
                {
                    channel.QueueDeclare(queue: consumeInterface.QueueName(),
                                         durable: true,
                                         exclusive: false,
                                         autoDelete: false,
                                         arguments: null);

                    channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);

                    Console.WriteLine(" [*] Waiting for messages.");
                    //保证在接收端一个消息没有处理完时不会接收另一个消息，即接收端发送了ack后才会接收下一个消息。在这种情况下发送端会尝试把消息发送给下一个not busy的接收端
                    var consumer = new EventingBasicConsumer(channel);
                    consumer.Received += (sender, ea) =>
                    {
                        var body = ea.Body.ToArray();

                        consumeInterface.ConsumeEvent(body);

                        // Note: it is possible to access the channel via
                        //       ((EventingBasicConsumer)sender).Model here
                        channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
                    };
                    channel.BasicConsume(queue: consumeInterface.QueueName(),
                                         autoAck: false,
                                         consumer: consumer);

                }
            });


        }
        /// <summary>
        /// 关闭
        /// </summary>
        public void Close()
        {
            if (_connection != null)
            {
                _connection.Close();
            }
        }
    }
}
